package com.usian.article.listener;

import com.usian.common.constants.message.HotArticleConstants;
import com.usian.common.util.JsonUtils;
import com.usian.model.article.msg.ArticleBehaviorMsg;
import com.usian.model.article.msg.ArticleBehaviorStreamMsg;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;

/**
 * 文章分值计算的监听类
 */
@EnableBinding(HotArticleStreamListener.HotArticleProcessor.class)
public class HotArticleStreamListener {

    @StreamListener(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC)
    @SendTo(HotArticleConstants.HOT_ARTICLE_SCORE_CONSUMER_TOPIC)
    public KStream<String, String> process(KStream<String, String> input){
        return input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            //            value的格式 {"articleId":12312312,"type":1,"num":1}
            @Override
            public KeyValue<String, String> apply(String key, String value) {
//                把value转对象
                ArticleBehaviorMsg articleBehaviorMsg = JsonUtils.toBean(value, ArticleBehaviorMsg.class);
                return new KeyValue<>(articleBehaviorMsg.getArticleId().toString(), value);
            }
        })
                .groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
                .windowedBy(TimeWindows.of(Duration.ofSeconds(15)))
                .aggregate(
                        new Initializer<String>() {
                            @Override
                            public String apply() {
                                return null;
                            }
                        },
                        new Aggregator<String, String, String>() {
                            //                            key-articleId, value - 其他服务传过来的消息{"articleId":12312312,"type":1,"num":1}  aggregate- 上一次计算的结果
                            @Override
                            public String apply(String key, String value, String aggregate) {
                                ArticleBehaviorStreamMsg streamMsg = null;
//                                如果上一次计算结果是否为空，创建新的ArticleBehaviorStreamMsg对象
                                if (StringUtils.isBlank(aggregate)) {
                                    streamMsg = new ArticleBehaviorStreamMsg();
                                    streamMsg.setArticleId(Long.valueOf(key));
                                    streamMsg.setCollection(0);
                                    streamMsg.setComment(0);
                                    streamMsg.setLikes(0);
                                    streamMsg.setViews(0);
                                } else {
//                                如果上一次计算结果不为空，转ArticleBehaviorStreamMsg对象
                                    streamMsg = JsonUtils.toBean(aggregate, ArticleBehaviorStreamMsg.class);
                                }
//                                把value转对象
                                ArticleBehaviorMsg articleBehaviorMsg = JsonUtils.toBean(value, ArticleBehaviorMsg.class);
                                System.out.println("####behaviorMsg="+articleBehaviorMsg);
                                switch (articleBehaviorMsg.getType()) {
                                    case LIKES:
                                        streamMsg.setLikes(streamMsg.getLikes() + articleBehaviorMsg.getNum());
                                        break;
                                    case VIEWS:
                                        streamMsg.setViews(streamMsg.getViews() + articleBehaviorMsg.getNum());
                                        break;
                                    case COMMENT:
                                        streamMsg.setComment(streamMsg.getComment() + articleBehaviorMsg.getNum());
                                        break;
                                    case COLLECTION:
                                        streamMsg.setCollection(streamMsg.getCollection() + articleBehaviorMsg.getNum());
                                        break;
                                }
                                System.out.println("streamMsg=="+streamMsg);
                                return JsonUtils.toString(streamMsg);
                            }
                        }
                        , Materialized.with(Serdes.String(), Serdes.String()))
                .toStream()
                .map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> key, String value) {
                        return new KeyValue<>(key.key(), value);
                    }
                });
    }

    public interface HotArticleProcessor {
        //输入消息
        @Input(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC)
        KStream<String, String> input();

        //输出消息
        @Output(HotArticleConstants.HOT_ARTICLE_SCORE_CONSUMER_TOPIC)
        KStream<String, String> output();
    }
}
